package com.xbk.seckill.ratelimiter.concurrent.interceptor;

import cn.hutool.core.util.StrUtil;
import com.xbk.seckill.common.exception.SeckillException;
import com.xbk.seckill.ratelimiter.concurrent.annotation.ConcurrentRateLimiter;
import com.xbk.seckill.ratelimiter.concurrent.bean.BHConcurrentRateLimiter;
import com.xbk.seckill.ratelimiter.concurrent.policy.ConcurrentRateLimiterPolicy;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 并发数限流拦截器
 *
 * @author xiexu
 * @date 2025/05/07 14:45
 */
@Slf4j
@Aspect
@Component
@ConditionalOnProperty(prefix = "rate.limit.local.concurrent", name = "enabled", havingValue = "true")
public class ConcurrentRateLimiterInterceptor implements EnvironmentAware, DisposableBean {

    private static final Map<String, BHConcurrentRateLimiter> BH_CONCURRENT_RATE_LIMITER_MAP = new ConcurrentHashMap<>();

    private Environment environment;

    @Value("${rate.limit.local.concurrent.default.corePoolSize:3}")
    private int defaultCorePoolSize;

    @Value("${rate.limit.local.concurrent.default.maximumPoolSize:5}")
    private int defaultMaximumPoolSize;

    @Value("${rate.limit.local.concurrent.default.queueCapacity:10}")
    private int defaultQueueCapacity;

    @Value("${rate.limit.local.concurrent.default.keepAliveTime:30}")
    private long defaultKeepAliveTime;

    @Value("${rate.limit.local.concurrent.default.timeout:1}")
    private long defaultTimeOut;

    @Pointcut("@annotation(concurrentRateLimiter)")
    public void pointCut(ConcurrentRateLimiter concurrentRateLimiter) {

    }

    @Around(value = "pointCut(concurrentRateLimiter)")
    public Object around(ProceedingJoinPoint pjp, ConcurrentRateLimiter concurrentRateLimiter) throws Throwable {
        // 获取被拦截方法的基础信息
        MethodSignature signature = (MethodSignature) pjp.getSignature();
        // 目标类名
        String className = pjp.getTarget().getClass().getSimpleName();
        // 方法名
        String methodName = signature.getName();
        // 解析限流器名称（支持SpEL表达式和默认生成规则）
        String rateLimitName = environment.resolvePlaceholders(concurrentRateLimiter.name());
        // 处理未配置名称或占位符未解析的情况：默认格式为 "类名-方法名"
        if (StrUtil.isEmpty(rateLimitName) || rateLimitName.contains("${")) {
            rateLimitName = className + "-" + methodName;
        }
        // 获取或创建限流器实例（带缓存机制）
        BHConcurrentRateLimiter rateLimiter = this.getRateLimiter(rateLimitName, concurrentRateLimiter);
        Object[] args = pjp.getArgs();
        // 基于自定义线程池实现并发数限流
        return rateLimiter.submit(() -> {
            try {
                return pjp.proceed(args);
            } catch (Throwable e) {
                if (e instanceof SeckillException) {
                    throw (SeckillException) e;
                }
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 获取BHConcurrentRateLimiter对象
     */
    private BHConcurrentRateLimiter getRateLimiter(String rateLimitName, ConcurrentRateLimiter concurrentRateLimiter) {
        // 尝试先从Map缓存中获取已存在的限流器，避免重复创建
        BHConcurrentRateLimiter bhRateLimiter = BH_CONCURRENT_RATE_LIMITER_MAP.get(rateLimitName);
        // 双重检查锁（Double-Check Locking）解决并发创建问题
        if (bhRateLimiter == null) {
            // 使用字符串常量池中的唯一对象作为锁对象（避免不同限流器名称的锁冲突）
            final String finalRateLimitName = rateLimitName.intern();
            synchronized (finalRateLimitName) {
                // 再次检查缓存（防止其他线程已创建）
                bhRateLimiter = BH_CONCURRENT_RATE_LIMITER_MAP.get(rateLimitName);
                // 获取的bhRateLimiter再次为空
                if (bhRateLimiter == null) {
                    // 解析线程池参数（优先使用注解配置，否则取默认值）
                    // 核心线程数（corePoolSize）
                    int corePoolSize = concurrentRateLimiter.corePoolSize() <= 0 ? defaultCorePoolSize : concurrentRateLimiter.corePoolSize();
                    // 最大线程数（maximumPoolSize）
                    int maximumPoolSize = concurrentRateLimiter.maximumPoolSize() <= 0 ? defaultMaximumPoolSize : concurrentRateLimiter.maximumPoolSize();
                    // 任务队列容量（queueCapacity）
                    int queueCapacity = concurrentRateLimiter.queueCapacity() <= 0 ? defaultQueueCapacity : concurrentRateLimiter.queueCapacity();
                    // 空闲线程存活时间（keepAliveTime）
                    long keepAliveTime = concurrentRateLimiter.keepAliveTime() <= 0 ? defaultKeepAliveTime : concurrentRateLimiter.keepAliveTime();
                    TimeUnit keepAliveTimeUnit = concurrentRateLimiter.keepAliveTimeUnit();
                    // 任务提交超时时间（timeout）
                    long timeout = concurrentRateLimiter.timeout() <= 0 ? defaultTimeOut : concurrentRateLimiter.timeout();
                    TimeUnit timeoutUnit = concurrentRateLimiter.timeoutUnit();

                    // 创建自定义线程池（每个限流器独立线程池）
                    ThreadPoolExecutor executor = new ThreadPoolExecutor(
                            // 核心线程数
                            corePoolSize,
                            // 最大线程数
                            maximumPoolSize,
                            // 非核心线程空闲存活时间
                            keepAliveTime,
                            // 时间单位
                            keepAliveTimeUnit,
                            // 有界队列（控制堆积量）
                            new ArrayBlockingQueue<>(queueCapacity),
                            // 线程工厂（命名可辨识）
                            (r) -> new Thread(r, "rate-limiter-threadPool-".concat(rateLimitName).concat("-"))
                            // 拒绝策略
                            , new ConcurrentRateLimiterPolicy());

                    // 包装为限流器对象（绑定线程池和超时配置）
                    bhRateLimiter = new BHConcurrentRateLimiter(executor, timeout, timeoutUnit);
                    // 写入缓存（putIfAbsent防止并发覆盖）
                    BH_CONCURRENT_RATE_LIMITER_MAP.putIfAbsent(rateLimitName, bhRateLimiter);
                }
            }
        }
        return bhRateLimiter;
    }

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    @Override
    public void destroy() throws Exception {
        // 关闭线程池
        log.info("destroy|关闭线程池");
        Map<String, BHConcurrentRateLimiter> map = BH_CONCURRENT_RATE_LIMITER_MAP;
        if (map.size() > 0) {
            for (Map.Entry<String, BHConcurrentRateLimiter> entry : map.entrySet()) {
                BHConcurrentRateLimiter rateLimiter = entry.getValue();
                rateLimiter.shutdown();
            }
            map.clear();
            BH_CONCURRENT_RATE_LIMITER_MAP.clear();
        }
    }
}
